package com.shujia.spark.streaming

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo01WordCntOnStreaming {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local[2]") // 至少需要两个线程，因为接收数据需要一致占用一个线程
      .config("spark.default.parallelism", "2")
      .getOrCreate()

    // 构建StreamingContext：需要传入SparkContext以及Duration表示批次大小
    val ssc: StreamingContext = new StreamingContext(spark.sparkContext, Durations.seconds(5))

    // 通过Socket连接 从nc服务中加载数据
    // Socket服务端：nc -lk 8888
    // 会返回DStream类型，SparkStreaming的一种编程模型
    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)

    lineDS
      // 将每一行数据切分并展开
      .flatMap(_.split(","))
      // 给每个单词后面加上1
      .map(w => (w, 1))
      // 默认是无状态的计算，每个批次结果都是独立的，互不影响
      .reduceByKey(_ + _)
      .print()

    // 启动Streaming任务
    ssc.start()
    ssc.awaitTermination() // 等待程序中断

  }

}
